package org.wu.framework.lazy.orm.database.sql.expand.database.persistence.event;

import org.framework.wu.framework.queue.Message;
import org.framework.wu.framework.queue.MessageQueue;
import org.framework.wu.framework.queue.MessageQueueFactory;
import org.wu.framework.core.utils.DataTransformUntil;
import org.wu.framework.core.utils.ObjectUtils;
import org.wu.framework.lazy.orm.database.lambda.stream.LambdaTableType;
import org.wu.framework.lazy.orm.database.sql.expand.database.persistence.method.LazyOperationParameter;
import org.wu.framework.lazy.orm.database.sql.util.SqlUtils;

import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class LazyOrmEventConsumer {

    private final LazyOperationParameter lazyOperationParameter;

    private final List<LazyOrmEvent<?>> lazyOrmEvents;

    ThreadPoolExecutor LAZY_ORM_EVENT_EXECUTOR =
            new ThreadPoolExecutor(10, 20, 3L, TimeUnit.MINUTES,
                    new LinkedBlockingDeque<>(50), new ThreadPoolExecutor.AbortPolicy());


    public LazyOrmEventConsumer(LazyOperationParameter lazyOperationParameter, List<LazyOrmEvent<?>> lazyOrmEvents) {
        this.lazyOperationParameter = lazyOperationParameter;
        this.lazyOrmEvents = lazyOrmEvents;
    }


    public void run() throws Exception {
        String eventQueueName = lazyOperationParameter.getLazyOperationAttribute().getEventQueueName();
        MessageQueue queue = MessageQueueFactory.getQueue(eventQueueName);
        Map<String, ? extends LazyOrmEvent<?>> LazyOrmEvent_Cache_Map = lazyOrmEvents.stream().collect(Collectors.toMap(LazyOrmEvent::getEventTable, lazyOrmEvent -> lazyOrmEvent, (A, B) -> A));
        // 创建监听线程
        Thread thread = new Thread(() -> {
            while (true) {
                Message receive = queue.receive();
                if (ObjectUtils.isNotEmpty(receive)) {
                    Object sql = receive.getBody();
                    String topic = receive.getTopic();
                    List<String> tablesInSql = SqlUtils.tablesInSql(sql.toString());
                    // TODO通过sql解析数据为map
                    Map<String, Object> data = SqlUtils.dataInSql(sql.toString());
                    for (String tableName : tablesInSql) {

                        try {
                            if (LazyOrmEvent_Cache_Map.containsKey(tableName)) {
                                LazyOrmEvent<?> lazyOrmEvent = LazyOrmEvent_Cache_Map.get(tableName);
                                // 获取范型
                                ParameterizedType parameterizedType = (ParameterizedType) lazyOrmEvent.getClass().getGenericSuperclass();
                                Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
                                Type actualTypeArgument = actualTypeArguments[0];
                                if (actualTypeArgument instanceof ParameterizedType resultActualType && Map.class.isAssignableFrom((Class<?>) resultActualType.getRawType())) {
                                    actualTypeArgument = Map.class;
                                }
                                assert actualTypeArgument instanceof Class<?>;
                                Class<?> entityClass = (Class<?>) actualTypeArgument;
                                Object entity = DataTransformUntil.transformBean(data, entityClass);
                                if (LambdaTableType.INSERT.name().equals(topic)) {
                                    Method declaredMethod = lazyOrmEvent.getClass().getDeclaredMethod("insertEvent", entityClass);
                                    declaredMethod.invoke(lazyOrmEvent, entity);

                                } else if (LambdaTableType.DELETE.name().equals(topic)) {
                                    Method declaredMethod = lazyOrmEvent.getClass().getDeclaredMethod("updateEvent", entityClass);
                                    declaredMethod.invoke(lazyOrmEvent, entity);
                                }
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }


                    // 解析数据类型
                    // 解析数据
                }
            }

        });
        LAZY_ORM_EVENT_EXECUTOR.submit(thread::start);
    }
}
